package com.mars.kafkademo.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * author by: mars
 * Date: 2020/5/4 22:49
 * Description:
 */
public class CountInterceptor implements ProducerInterceptor<String,String> {

    private long successCount = 0;
    private long errorCount = 0;
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 不做拦截内容的改变，直接返回record
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            // 表示成功，可以将成功失败个数写入文件，可用bufferIO内存流，也可以在关闭资源的时候统计
            successCount++;
        } else {
            // 表示失败
            errorCount++;
        }
    }

    @Override
    public void close() {
        System.out.println("成功的个数："+ successCount);
        System.out.println("失败的个数："+ errorCount);
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
